import com.rabbitmq.client.*;
import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

@Slf4j
//负责从rabbitmq接受各种模式的消息
public class RabbitReceiver {
    static Connection conn;

    @BeforeAll
    public static void createConn() throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.56.77");
        factory.setPort(5672);
        factory.setUsername("rabbitmq");
        factory.setPassword("rabbitmq");
        factory.setVirtualHost("/mdmall");
        conn = factory.newConnection();
    }

    @Test
    public void r1() throws IOException, TimeoutException, InterruptedException {
        Channel channel = conn.createChannel();//建立信道
        String queueName = "simple_queue";
        channel.queueDeclare(queueName, true, false, true, null);//声明队列
        //创建一个默认的消费者实例，然后和前面声明的信道建立联系
        Consumer consumer = new DefaultConsumer(channel) {
            //当接收到新消息时会自动执行这个方法
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                log.info("1号消费者接收到新消息  {}", new String(body, StandardCharsets.UTF_8));
            }
        };
        /**
         * 设置消费者指定监听的队列
         * 参数    队列名称    是否自动确认，接收到消息之后自动向消息队列回复一个ack确认通知    消费者实例
         */
        channel.basicConsume(queueName, true, consumer);
        TimeUnit.MINUTES.sleep(2);
        channel.close();
        conn.close();
    }

    @Test
    public void r2() throws IOException, TimeoutException, InterruptedException {
        Channel channel = conn.createChannel();//建立信道
        String queueName = "work_queue";
        channel.queueDeclare(queueName, true, false, true, null);//声明队列
        Consumer c1 = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                log.info("1111111111111号消费者接收到新消息  ：{}", new String(body, StandardCharsets.UTF_8));
            }
        };
        log.info("111111111111号准备完毕");
        channel.basicConsume(queueName,true,c1);
        Consumer c2 = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                log.info("22222222222222号消费者接收到新消息  ：{}", new String(body, StandardCharsets.UTF_8));
            }
        };
        log.info("2号2222222222准备完毕");
        channel.basicConsume(queueName,true,c2);
        Consumer c3 = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                log.info("3333333333333号消费者接收到新消息  ：{}", new String(body, StandardCharsets.UTF_8));
            }
        };
        log.info("333333333333333号准备完毕");
        channel.basicConsume(queueName,true,c3);
        TimeUnit.MINUTES.sleep(2);
        channel.close();
        conn.close();
    }
    @Test  //扇形交换机，发布/订阅模式
    public void r3() throws IOException, TimeoutException, InterruptedException{
        Channel channel = conn.createChannel();
        String queue1 = "fanout_queue1",queue2 = "fanout_queue2";
        channel.queueDeclare(queue1,true,false,false,null);
        channel.queueDeclare(queue1,true,false,false,null);
        DefaultConsumer c5 = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                log.info("5555555555555号消费者接收到新消息：{}",new String(body,StandardCharsets.UTF_8));
            }
        };
        channel.basicConsume(queue1,c5);//订阅队列1
        DefaultConsumer c6 = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                log.info("666666666666666666666666号消费者接收到新消息：{}",new String(body,StandardCharsets.UTF_8));
            }
        };
        channel.basicConsume(queue2,c6);//订阅队列2
        log.info("555555555555号准备完毕1111111111111");
        log.info("666666666666号准备完毕2222222222222");
        TimeUnit.MINUTES.sleep(2);
        channel.close();
        conn.close();
    }
    @Test  //路由发布/订阅模式
    public void r4() throws IOException, TimeoutException, InterruptedException{
        Channel channel = conn.createChannel();
        String queue1 = "direct_queue1",queue2 = "direct_queue2";
        channel.queueDeclare(queue1,true,false,false,null);
        channel.queueDeclare(queue1,true,false,false,null);
        DefaultConsumer c5 = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                log.info("5555555555555号消费者接收到新消息：{}",new String(body,StandardCharsets.UTF_8));
            }
        };
        channel.basicConsume(queue1,c5);//订阅队列1
        DefaultConsumer c6 = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                log.info("666666666666666666666666号消费者接收到新消息：{}",new String(body,StandardCharsets.UTF_8));
            }
        };
        channel.basicConsume(queue2,c6);//订阅队列2
        log.info("555555555555号准备完毕1111111111111");
        log.info("666666666666号准备完毕2222222222222");
        TimeUnit.MINUTES.sleep(2);
        channel.close();
        conn.close();
    }
    @Test  //主题模式发布/订阅模式
    public void r5() throws IOException, TimeoutException, InterruptedException{
        Channel channel = conn.createChannel();
        String queue1 = "topic_queue1",queue2 = "topic_queue2";
        channel.queueDeclare(queue1,true,false,false,null);
        channel.queueDeclare(queue1,true,false,false,null);
        DefaultConsumer c5 = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                log.info("5555555555555号消费者接收到新消息：{}",new String(body,StandardCharsets.UTF_8));
            }
        };
        channel.basicConsume(queue1,c5);//订阅队列1
        DefaultConsumer c6 = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                log.info("666666666666666666666666号消费者接收到新消息：{}",new String(body,StandardCharsets.UTF_8));
            }
        };
        channel.basicConsume(queue2,c6);//订阅队列2
        log.info("555555555555号准备完毕1111111111111");
        log.info("666666666666号准备完毕2222222222222");
        TimeUnit.MINUTES.sleep(2);
        channel.close();
        conn.close();
    }
}
